package spring.cloud.tasks.api.service.impl;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import spring.cloud.tasks.api.database.entity.NameSpaceZooKeeperClusterMappingDto;
import spring.cloud.tasks.api.database.repository.NameSpaceZooKeeperClusterMappingDtoRepository;
import spring.cloud.tasks.api.domain.NameSpace;
import spring.cloud.tasks.api.service.NameSpaceAndTaskService;
import spring.cloud.tasks.api.service.NameSpaceService;
import spring.cloud.tasks.api.service.ZooKeeperService;
import spring.cloud.tasks.api.utils.SaturnThreadFactory;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class NameSpaceAndTaskServiceImpl implements NameSpaceAndTaskService {

    private static final Logger logger = LoggerFactory.getLogger(NameSpaceAndTaskServiceImpl.class);

    @Autowired
    private ZooKeeperService zooKeeperService;

    @Autowired
    private NameSpaceZooKeeperClusterMappingDtoRepository namespaceZooKeeperClusterMappingRepository;

    @Autowired
    private NameSpaceService namespaceService;

    private ExecutorService executorService;

    @PostConstruct
    public void init() {
        if (executorService == null) {
            executorService = Executors.newFixedThreadPool(5, new SaturnThreadFactory("Saturn-NamespaceAndJob-Thread"));
        }
    }

    @PreDestroy
    public void destroy() {
        if (executorService != null) {
            executorService.shutdownNow();
        }
    }

    @Override
    public void createNameSpaceAndCloneTasks(String srcNameSpace, String nameSpace, String zooKeeperClusterKey, String createBy) throws RuntimeException {
        logger.info("start createNamespaceAndCloneJobs, srcNamespace: {}, namespace: {}, zkClusterName: {}", srcNameSpace, nameSpace, zooKeeperClusterKey);
        NameSpaceZooKeeperClusterMappingDto dto = namespaceZooKeeperClusterMappingRepository.selectByNameSpace(srcNameSpace);
        if (dto == null) {
            throw new RuntimeException("no zkCluster mapping is not found");
        }
        NameSpace namespaceInfo = new NameSpace();
        namespaceInfo.setNameSpace(nameSpace);
        namespaceInfo.setZooKeeperClusterKey(zooKeeperClusterKey);
        namespaceInfo.setContent("");
        try {
            zooKeeperService.createNameSpace(namespaceInfo);
            zooKeeperService.refreshRegistryCenterForNamespace(zooKeeperClusterKey, srcNameSpace);
        } catch (RuntimeException e) {
            if (StringUtils.equals(String.format(ZooKeeperServiceImpl.ERR_MSG_NS_ALREADY_EXIST, nameSpace), e.getMessage())) {
                logger.warn("namespace already exists, ignore this exception and move on");
            } else {
                throw e;
            }
        }
        namespaceService.importTasksFromNameSpaceToNameSpace(srcNameSpace, nameSpace, createBy);
        logger.info("finish createNamespaceAndCloneJobs, srcNamespace: {}, namespace: {}, zkClusterName: {}", srcNameSpace, nameSpace, zooKeeperClusterKey);
    }

    @Override
    public void asyncCreateNameSpaceAndCloneTasks(final String srcNamespace, final String nameSpace,
                                                  final String zkClusterName, final String createBy) {
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    createNameSpaceAndCloneTasks(srcNamespace, nameSpace, zkClusterName, createBy);
                } catch (RuntimeException e) {
                    logger.warn("fail to create and clone jobs, srcNamespace: {}, namespace: {}, zkClusterName: {}", srcNamespace, nameSpace, zkClusterName, e);
                }
            }
        });
    }
}
